package com.heima.article.steam;


import com.alibaba.fastjson.JSON;
import com.heima.common.constants.HotArticleConstants;
import com.heima.model.message.ArticleVisitStreamMess;
import com.heima.model.message.UpdateArticleMess;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;
import java.util.Arrays;

@Configuration
public class HotArticleStreamHandler {


    @Bean
    public KStream<String, String> process(StreamsBuilder streamsBuilder){
        KStream<String, String> kStream = streamsBuilder.stream(HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC);
        kStream.map((key,value)->{

            UpdateArticleMess updateArticleMess = JSON.parseObject(value, UpdateArticleMess.class);
            String type = updateArticleMess.getType().name();
//                 例如：   LIKES:0
            return new KeyValue<>(updateArticleMess.getArticleId().toString(),  type + ":" + updateArticleMess.getAdd());
        }).groupBy((key,value)->key)
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
                /**
                 * 自行的完成聚合的计算
                 */
                .aggregate(new Initializer<String>() {

                    /**
                     * 初始方法，返回值是消息的value
                     *
                     * @return
                     */
                    @Override
                    public String apply() {
                        return "COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0";
                    }
                }, new Aggregator<String, String, String>() {
                    /**
                     *
                     * @param key
                     * @param value  .map后的消息key
                     * @param aggValue  循环后的值："COLLECTION:5,COMMENT:4,LIKES:2,VIEWS:0"
                     * @return
                     */
                    @Override
                    public String apply(String key, String value, String aggValue) {
                        //初始值
                        int collection = 0; //收藏
                        int comment = 0; //评论
                        int likes = 0; //点赞
                        int views = 0; //浏览
                        String[] split = aggValue.split(",");
                        System.out.println("分割后数组" + Arrays.toString(split));
                        for (String typeAgg : split){
                            switch ( typeAgg.split(":")[0]){
                                case "COLLECTION":
                                    collection = Integer.parseInt(typeAgg.split(":")[1]);
                                    break;
                                case "COMMENT":
                                    comment = Integer.parseInt(typeAgg.split(":")[1]);
                                    break;
                                case "LIKES":
                                    likes = Integer.parseInt(typeAgg.split(":")[1]);
                                    break;
                                case "VIEWS":
                                    views = Integer.parseInt(typeAgg.split(":")[1]);
                            }
                        }
                        //赋值
                        String type = value.split(":")[0];
                        switch (type){
                            case "COLLECTION":
                                collection = collection + Integer.parseInt(value.split(":")[1]);
                                break;
                            case "COMMENT":
                                comment = comment + Integer.parseInt(value.split(":")[1]);
                                break;
                            case "LIKES":
                                likes = likes + Integer.parseInt(value.split(":")[1]);
                                break;
                            case "VIEWS":
                                views = views + Integer.parseInt(value.split(":")[1]);
                                break;
                        }
                        System.out.println("文章的id:"+key);
                        System.out.println("当前时间窗口内的消息处理结果：" + "COLLECTION:" + collection + ",COMMENT:" + comment + ",LIKES:" + likes + ",VIEWS:" + views);

                        return "COLLECTION:" + collection +",COMMENT:" + comment + ",LIKES:" + likes + ",VIEWS:" + views;
                    }
                }, Materialized.as("hot-atricle-stream-count-001"))
                .toStream()
                .map((key,value)->

                    new KeyValue<>(key.key(), convertFormat(key,value))

                )
                .to(HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC);

        return kStream;

    }

    private  String convertFormat(Windowed<String> key, String value) {
        String[] split = value.split(",");
        ArticleVisitStreamMess articleVisitStreamMess = new ArticleVisitStreamMess();
        articleVisitStreamMess.setArticleId(Long.valueOf( key.key()));
        for (String type : split){
            if (type.split(":")[0].equals("COLLECTION")){
                articleVisitStreamMess.setCollect(Integer.parseInt(type.split(":")[1]));
            }else if (type.split(":")[0].equals("COMMENT")){
                articleVisitStreamMess.setComment(Integer.parseInt(type.split(":")[1]));
            }else if (type.split(":")[0].equals("LIKES")){
                articleVisitStreamMess.setLike(Integer.parseInt(type.split(":")[1]));
            }else if (type.split(":")[0].equals("VIEWS")){
                articleVisitStreamMess.setView(Integer.parseInt(type.split(":")[1]));
            }
        }

        return JSON.toJSONString(articleVisitStreamMess);
    }
}
